package com.rem.flink.flink7Stream;

import com.rem.flink.flink2Source.ClickSource;
import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 对流的操作
 * 分流 根据name 分流 输出到侧输出流
 *
 * @author Rem
 * @date 2022-10-17
 */

public class SplitStreamByOutPutTag {


    public static void main(String[] args) throws Exception {
        // 定义输出标签，侧输出流的数据类型为三元组(user, url, timestamp)
        OutputTag<Tuple3<String, String, Long>> maryTag = new OutputTag<>("Mary-pv") {
        };
        OutputTag<Tuple3<String, String, Long>> aliceTag = new OutputTag<>("Bob-pv") {
        };


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource());

        SingleOutputStreamOperator<Event> process = stream.process(new ProcessFunction<Event, Event>() {
            @Override
            public void processElement(Event event, ProcessFunction<Event, Event>.Context ctx, Collector<Event> out) throws Exception {
                if ("Mary".equals(event.getUser())) {
                    ctx.output(maryTag, new Tuple3<>(event.getUser(), event.getUrl(), event.getTimestamp()));
                } else if ("Alice".equals(event.getUser())) {
                    ctx.output(aliceTag, new Tuple3<>(event.getUser(), event.getUrl(), event.getTimestamp()));
                } else {
                    out.collect(event);
                }
            }
        });

        process.getSideOutput(maryTag).print("Mary pv");
        process.getSideOutput(aliceTag).print("Alice pv");
        process.print("other pv");

        env.execute();
    }
}
